package com.javaspeak.concurrency.nio.niolet.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.nio.charset.CharacterCodingException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.LockSupport;

import org.apache.log4j.Logger;


/**
 * @author John Dickerson
 */
public class ServerImpl extends Thread implements Server {

    public Logger logger = Logger.getLogger( ServerImpl.class );
    private volatile boolean keepRunning = true;
    private final int defaultBufferSize;
    private final int port;
    private final int numberReadProcessingThreads;
    private Selector socketSelector;
    private ServerSocketChannel serverSocketChannel;

    private ConcurrentLinkedQueue<Message> messageConcurrentLinkedQueue =
        new ConcurrentLinkedQueue<Message>();

    private ConcurrentLinkedQueue<Message>
        messageResponseConcurrentLinkedQueue =
            new ConcurrentLinkedQueue<Message>();

    private final Map<SelectionKey, ByteBuffer> readBufferBySelectionKeyMap =
        new ConcurrentHashMap<SelectionKey, ByteBuffer>();

    private ServerNiolet serverNiolet;

    private MessageProcessingThread[] messageProcessingThreads;

    private MessageResponseThread messageResponseThread;


    /**
     * Initializes Server
     *
     * @param hostInetAddress
     * @param port
     * @throws IOException
     */
    private void initializeServer()
        throws ServerException {

        try {
            // Create a new selector
            socketSelector = SelectorProvider.provider().openSelector();

            // Create a new non-blocking server socket channel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking( false );

            InetAddress localHostInetAddress = InetAddress.getLocalHost();

            logger.debug( "Server hostName: " +
                    localHostInetAddress.getHostAddress() );

            // Bind the server socket to the specified address and port
            InetSocketAddress inetSocketAddress =
                new InetSocketAddress( localHostInetAddress, this.port );

            serverSocketChannel.socket().bind( inetSocketAddress, 5000 );

            // Register the server socket channel, indicating an interest in
            // accepting new connections
            serverSocketChannel.register(
                    socketSelector, SelectionKey.OP_ACCEPT );

            messageProcessingThreads =
                new MessageProcessingThread[ numberReadProcessingThreads ];


            messageResponseThread =
                new MessageResponseThread(
                        messageResponseConcurrentLinkedQueue,
                            readBufferBySelectionKeyMap );

            for ( int i=0; i<numberReadProcessingThreads; i++ ){

                messageProcessingThreads[ i ] = new MessageProcessingThread(
                        messageConcurrentLinkedQueue,
                            messageResponseConcurrentLinkedQueue,
                                messageResponseThread,
                                    serverNiolet  );

                messageProcessingThreads[ i ].start();
            }

            messageResponseThread.start();
        }
        catch( UnknownHostException e ){

            throw new ServerException( "Could not initialize Server", e );
        }
        catch( ClosedChannelException e ){

             throw new ServerException( "Could not initialize Server", e );
        }
        catch( IOException e ){

            throw new ServerException( "Could not initialize Server", e );
        }
    }


    private void connect( SelectionKey selectionKey ) throws IOException{

        SocketChannel socketChannel =
            ( SocketChannel )selectionKey.channel();

        socketChannel.finishConnect();
    }


    private void accept( SelectionKey selectionKey ) throws IOException {

        ServerSocketChannel serverSocketChannel
            = ( ServerSocketChannel )selectionKey.channel();

        // Accept the connection and make it non-blocking
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking( false );
        socketChannel.register( socketSelector, SelectionKey.OP_READ );
    }


    private void read( SelectionKey selectionKey ) throws IOException {

        SocketChannel socketChannel =
            ( SocketChannel )selectionKey.channel();

        ByteBuffer readBuffer =
            readBufferBySelectionKeyMap.get( selectionKey );

        if ( readBuffer == null ){

            readBuffer = ByteBuffer.allocate( defaultBufferSize );
            readBufferBySelectionKeyMap.put( selectionKey, readBuffer );
        }

        if ( socketChannel.read( readBuffer ) == -1 ){

            throw new IOException( "Read on closed SelectionKey" );
        }

        /**
        if ( logger.isDebugEnabled() ){

            debugByteBufferInWriteMode( readBuffer );
        }
        **/

        // Get ready for reading data from byte array
        readBuffer.flip();

        ByteBuffer messageByteBuffer;

        while ( ( messageByteBuffer =
            readMessage( selectionKey, readBuffer ) ) != null  ){

            addMessageToMessageQueue( messageByteBuffer, selectionKey );

            for ( MessageProcessingThread messageProcessingThread :
                messageProcessingThreads ){

                LockSupport.unpark( messageProcessingThread );
            }
        }
    }


    /**
     * Capacity: Inside the ByteBuffer, there is a backing byte[] or
     * something that behaves much like one. The capacity is its size. The
     * capacity indexes the first slot past the end of the buffer.
     *
     * Limit: When filling the buffer, the limit is the same as the capacity.
     * When emptying the buffer, it is one past the last filled byte in the
     * buffer.
     *
     * Position: When filling the buffer, the position points just past the
     * last byte filled in the buffer. When emptying the buffer, the position
     * points just past the last byte written from the buffer.
     *
     * @param selectionKey
     * @param readBuffer
     * @return
     */
    private ByteBuffer readMessage(
            SelectionKey selectionKey, ByteBuffer readBuffer ){

        int bytesToRead;

        // test whether buffer has 4 bytes. The first 4 bytes are the size of
        // the rest of the message after the 4 bytes
        if ( readBuffer.remaining() > 4 ) {

            // read number of bytes in rest of message
            bytesToRead = readBuffer.getInt();

            // Test whether need to continue waiting for new bytes to be written
            if ( ( readBuffer.limit() - readBuffer.position() ) < bytesToRead ) {

                // if the buffer is full up resize it
                if ( readBuffer.limit() == readBuffer.capacity() ) {

                    int oldCapacity = readBuffer.capacity();

                    ByteBuffer tmp = ByteBuffer.allocate( bytesToRead + 4 );
                    readBuffer.position( 0 );
                    tmp.put( readBuffer );
                    readBuffer = tmp;
                    readBuffer.position( oldCapacity );
                    readBuffer.limit( readBuffer.capacity() );
                    readBufferBySelectionKeyMap.put( selectionKey, readBuffer );
                    return null;
                }
                else {
                    // rest for writing
                    readBuffer.position( readBuffer.limit() );
                    readBuffer.limit( readBuffer.capacity() );
                    return null;
                }
            }
        } else {
            // do not have 4 bytes yet. Continue waiting for enough bytes to
            // be written

            readBuffer.position( readBuffer.limit() );
            readBuffer.limit( readBuffer.capacity() );
            return null;
        }

        // read bytes into an array
        byte[] resultMessage = new byte[ bytesToRead ];
        readBuffer.get( resultMessage, 0, bytesToRead );

        // remove read message from buffer. Any further bytes in the buffer
        // are for more messages
        int remaining = readBuffer.remaining();
        readBuffer.limit( readBuffer.capacity() );
        readBuffer.compact();
        readBuffer.position( 0 );
        readBuffer.limit( remaining );

        return ByteBuffer.wrap( resultMessage );
    }

    /**
    private void debugByteBufferInWriteMode( ByteBuffer byteBuffer ){

        logger.debug( "WRITING: position = " + byteBuffer.position() +
                    " limit = " + byteBuffer.limit() +
                        " capacity = " + byteBuffer.capacity() + " remaining = " +
                            byteBuffer.remaining() );


    }
    **/


    private void addMessageToMessageQueue(
            ByteBuffer messageReceivedByteBuffer, SelectionKey selectionKey )
                throws CharacterCodingException{

        messageConcurrentLinkedQueue.add(
                new Message( messageReceivedByteBuffer, selectionKey ) );
    }


    /**
     * Remove any internal state for the given key.
     * @param key the key to be cancelled.
     */
    private void resetKey( SelectionKey selectionKey ) {

        selectionKey.cancel();
        readBufferBySelectionKeyMap.remove( selectionKey );
    }


    /**
     * Callback method for when the server disconnects an client connection.
     * @param key the key for the disconnected client.
     */
    public void disconnected( SelectionKey selectionKey ){

        // logger.debug( "Disconnecting client" );
    }


    /**
     * Not implemented
     *
     * @param selectionKey
     */
    public void write( SelectionKey selectionKey ){

    }


    /**
     * Constructor
     *
     * @param hostAddress
     * @param port
     * @throws IOException
     */
    public ServerImpl(
            int port, int defaultBufferSize, int numberReadProcessingThreads,
                ServerNiolet serverNiolet )
                    throws ServerException{

        this.defaultBufferSize = defaultBufferSize;
        this.port = port;
        this.numberReadProcessingThreads = numberReadProcessingThreads;
        this.serverNiolet = serverNiolet;
    }



    public void stopGracefully(){

        keepRunning = false;

        for ( int i=0; i<messageProcessingThreads.length; i++ ){

            messageProcessingThreads[ i ].gracefullyStop();
            LockSupport.unpark( messageProcessingThreads[ i ] );
        }

        messageResponseThread.gracefullyStop();
        LockSupport.unpark( messageResponseThread );
    }


    public void stopAggressively(){

        keepRunning = false;

        for ( int i=0; i<messageProcessingThreads.length; i++ ){

            if ( messageProcessingThreads[ i ] != null &&
                    ! messageProcessingThreads[ i ].isAlive() ){

                messageProcessingThreads[ i ].interrupt();
            }
        }

        if ( messageResponseThread != null &&
                    ! messageResponseThread.isAlive() ){

            messageResponseThread.interrupt();
        }
    }


    public void run() {

        try {
            initializeServer();

            while ( keepRunning ){
                // wait for an event on registered channels
                socketSelector.select();

                Iterator<SelectionKey> selectionKeys =
                    this.socketSelector.selectedKeys().iterator();

                while ( selectionKeys.hasNext()) {

                    SelectionKey selectionKey = null;

                    try {
                        selectionKey = selectionKeys.next();
                        selectionKeys.remove();

                        if ( ! selectionKey.isValid() ) {

                            continue;
                        }

                        if ( selectionKey.isConnectable() ){

                            // connect( selectionKey );
                        }
                        else if ( selectionKey.isAcceptable() ) {

                            accept( selectionKey );
                        }
                        else if ( selectionKey.isReadable() ){

                            read( selectionKey );
                        }
                        else if ( selectionKey.isWritable() ){

                            // not used because use a separate thread for
                            // writing called MessageResponseThread
                            write( selectionKey );
                        }
                    }
                    catch( IOException e ){

                        if ( selectionKey != null ){

                            resetKey( selectionKey );
                            disconnected( selectionKey );
                        }
                    }
                }
            }
        }
        catch( Throwable e ){

            throw new RuntimeException( "Server Failure", e );
        }
        finally {

            try {
                socketSelector.close();
                serverSocketChannel.socket().close();
            }
            catch( IOException e ){
                // do nothing
            }
        }
    }
}
